/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.mapreduce.example;

import org.sss.client.DataPutter;
import org.sss.client.SssClient;
import org.sss.mapreduce.GroupID;
import org.sss.mapreduce.HashPartitioner2;
import org.sss.mapreduce.datatype.DCPackableBytes;
import org.sss.mapreduce.datatype.PackableBytes;

public class PutData {

  public static void usage() {
    System.out.println("java PutData keysize keycard keycount valsize");
    System.out.println("  keysize must be >= 4");
    System.exit(-1);
  }

  /**
   * Usage:
   * java PutData keysize keycard keycount valsize
   */
  public static void main(String[] args) throws Exception {
    SssClient client = new SssClient(args);

    String[] others = client.getArgs();
    if (others.length != 4 || Integer.parseInt(others[0]) < 4) {
      usage();
    }
    int keysize = Integer.parseInt(others[0]);
    int keycard = Integer.parseInt(others[1]);
    int keycount = Integer.parseInt(others[2]);
    int valsize = Integer.parseInt(others[3]);

    DataPutter<DCPackableBytes, PackableBytes> putter =
      DataPutter.create(client, DCPackableBytes.class, PackableBytes.class,
                        HashPartitioner2.class);
    GroupID table = putter.getOutputGroupID();
    byte[] val = new byte[valsize];
    for (int i=0; i<valsize; i++) {
      val[i] = 0;
    }
    byte[] key = new byte[keysize];
    for (int i=0; i<keysize; i++) {
      key[i] = 0;
    }
    try {
      for (int card=0; card<keycard; card++) {
        key[0] = (byte) ( card        & 0xff);
        key[1] = (byte) ((card >>  8) & 0xff);
        key[2] = (byte) ((card >> 16) & 0xff);
        key[3] = (byte) ((card >> 24) & 0xff);
        for (int node=0; node<16; node++) {
          for (int slot=0; slot<16; slot++) {
            for (int i=0; i<keycount; i++) {
              putter.append(new DCPackableBytes(key, node, slot),
                            new PackableBytes(val));
            }
          }
        }
      }
    }
    finally {
      putter.close();
    }

    System.out.println("output data deployed - " + table.toString());
  }
}
